package com.geintech.learnobj;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

@Slf4j
@Controller
public class RocketMqConsumer {

    @Resource
    private RedisTemplate redisTemplate;

    @PostConstruct
    public void consume() {
        //创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");

        //NameServer地址，多个地址以;隔开
        consumer.setNamesrvAddr("rockmqserv:9876");
        try {
            //订阅名为的firstTopic的topic消息，且tag为first
            consumer.subscribe("topic", "tag");

            //如果第一次启动该消费者，那么从头开始消费
            //如果非第一次启动，那么按照上次消费的位置继续消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            /*
             * 注册一个并行消息监听器
             */
            consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                try {
                    System.out.println("size: " + list.size());
                    for (MessageExt messageExt : list) {

                        String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);

                        String[] split = StringUtils.split(messageBody, ",");
                        redisTemplate.opsForValue().set(split[0],split[1]);
                        System.out.println("new register user: " + messageBody);
                    }
                } catch (Exception e) {
                    log.error("error", e);
                    //并行失败&稍后重试
                    //实际会放入重试队列，默认最大重试16次，默认从level3的延迟开始，最终还是失败则存入死信队列
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            //启动消费者
            consumer.start();
        } catch (Exception e) {
            log.error("error", e);
            consumer.shutdown();
        }
    }

}
